身為一個佛系投資人，最重要的就是排程了，讓電腦不只會挑土豆，還會挑股票，今天介紹的是 Python 中蠻好用的一個套件 APScheduler - Advnaced Python Scheduler。APScheduler 有四個主要的組件

• triggers
• job stores
• executors
• schedulers

trigger

觸發器，分為 date, interval, cron 三種，date 就是指定時間執行一次，interval 就是隔多久執行一次，cron 就是 linux 的排程方式，cron 可針對 分, 時, 日, 月, 星期幾 去做設定何時去進行排程

job store

儲存工作的地方，預設是存在記憶體中，也可以存在資料庫裡，我們這次就是配合 SqlAlchemyJobStore 存在 Sqlite 的資料庫中。

executor

如何執行排程的工作，一般就是 thread pool 或是 process pool，一般來說如果是 IO 偏重的工作選用 thread pool, 如果是 CPU 偏重的工作，選用 process pool。

scheduler

整合以上三種的程式，一個程式在執行，一定要有一個主程式，這個 scheduler 就是主程式，主要有分以下幾種

• BlockingScheduler: 如果只有執行 APScheduler 就使用這一個
• BackgroundScheduler: 附屬在某個主程式底下，而且沒有使用以下的框架
• AsyncIOScheduler: 搭配 asyncio module
• GeventScheduler: 搭配 gevent
• TornadoScheduler: 搭配 Tornado
• TwistedScheduler: 搭配 Twisted
• QtScheduler: 搭配 Qt

安裝

pip install apscheduler

以下就簡單一個程式來啟動一個排程工作

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timedelta

def alarm(time):
    print("現在時間 %s" % time)
    
scheduler = BlocingScheduler()

scheduler.add_jobstore("sqlalchemy", url="sqlite:///schedule.sqlite3")

alarm_time = datetime.now() + timedelta(seconds=10)

scheduler.add_job(alarm, 'date', run_date = alarm_time, args=[datetime.now()])

scheduler.start()

試著跑跑看，應該可以顯示出時間來，接下來就是我們程式的修改，我們希望能夠每天下午 2 點半自動去回測最新的資料，如果有買入的訊號，用 line 通知我們次日進行買入，規劃的目錄結構如下:

project
│   main.py
│   jobs.py
│   strategies.py
│   services.py
│
└───repository
│   │   databse.py
│   │   models.py
│   │   database.sqlite3 (資料庫檔案)
│   │   api.py (shioaji api)

services.py

放置存取資料相關的程式，程式碼和昨天一樣，不過今天多加了一個 checkOffDay 檢查當天有沒有休市，如果是休市的話就不用執行了

def checkOffDay(day):
    db = session()
    
    result = db.query(OffDays).filter(OffDays.date == day).count()
    
    return result != 0

strateties.py

基本上就是放自己要測試的策略，這邊以 Day25 的葛蘭碧八法來示範，程式碼請參考之前

jobs.py

這邊就是放我們要執行的工作，相關程式碼如下:

import backtrader as bt
from datetime import timedelta, date
from services import checkOffDay, getKbars

def checkStrategy(stockList, strategy):
    """
    stockList: 傳入要監看的股票代碼
    strategy: 要執行的策略
    """
    for stock in stockList:
        
        today = date.today()
        
        # 檢查今天是不是假日
        if not checkOffDay(today):

            cerebro = bt.Cerebro()

            cerebro.addstrategy(strategy)

            # 因為回策所需的資料只有半年，目前設定 200 天，也可以設長一點
            # 最多就是第一次抓資料比較慢而已，之後會從本機的資料庫抓
            df = getKbars(stock, (today - timedelta(days=200)), today)
            
            data = bt.feeds.PandasData(dataname=df, timeframe=bt.TimeFrame.Minutes)

            cerebro.resampledata(data, timeframe=bt.TimeFrame.Days)
            
            cerebro.run()

main.py

主要執行的程式碼，主要就是建立一個 cron 的排程，每天下午 2 點執行回測，把要回測的股票代號放在 stocks 的陣列中，當做參數傳遞進去。如果 cron 不知道要怎麼設可以參考 Crontab guru 這個工具網站

from jobs import checkStrategy
from repository import models
from repository.database import engine
from apscheduler.schedulers.blocking import BlockingScheduler
from strategies import RuleOfEight

models.Base.metadata.create_all(bind=engine)

if __name__ == "__main__":
    stocks = ["2303", "2330", "2412", ]

    scheduler = BlockingScheduler()

    scheduler.add_job(checkStrategy, "cron", hour=14, minute=30, args=[stocks, RuleOfEight])
    
    scheduler.start()
    

執行的話，就直接用 python 執行 main.py

python main.py

執行完後，視窗不要關掉，因為 APScheduler 目前是在跑 BlockingScheduler 模式，一關掉就沒有了，明天我們再來介紹怎麼通知